package com.lzq.mq.nameserver.core;

import com.lzq.mq.common.coder.TcpMsgDecoder;
import com.lzq.mq.common.coder.TcpMsgEncoder;
import com.lzq.mq.common.constants.TcpConstants;
import com.lzq.mq.nameserver.common.CommonCache;
import com.lzq.mq.nameserver.enums.ReplicationModeEnum;
import com.lzq.mq.nameserver.enums.ReplicationRoleEnum;
import com.lzq.mq.common.event.EventBus;
import com.lzq.mq.nameserver.event.handler.TcpNettyServerHandler;
import com.lzq.mq.nameserver.replication.MasterReplicationMsgSendTask;
import com.lzq.mq.nameserver.replication.ReplicationService;
import com.lzq.mq.nameserver.replication.ReplicationTask;
import com.lzq.mq.nameserver.replication.SlaveReplicationHeartBeatTask;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import lombok.extern.slf4j.Slf4j;


@Slf4j
public class NameServerStarter {

    private final int port;

    private static ReplicationService replicationService;

    public NameServerStarter(int port) {
        this.port = port;
    }

    /**
     * 启动NameServer
     */
    public void startServer() throws InterruptedException {

        // 初始化消息总线
        EventBus.init();

        // 先启动复制任务, 后开启监听, 防止子节点丢失数据
        initReplication();

        // 启动超时检测任务
        startTimeoutRemovalTask();

        // 构建netty服务
        // 注入编解码器
        // 注入特定的handler
        // 启动netty服务
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();   // 处理网络io中的accept事件(接受连接)
        NioEventLoopGroup workerGroup = new NioEventLoopGroup(); // 处理网络io中的read/write事件

        ServerBootstrap bootstrap = new ServerBootstrap();       // 服务启动类
        bootstrap.group(bossGroup, workerGroup);                 // 绑定线程组
        bootstrap.channel(NioServerSocketChannel.class);         // 绑定服务通道

        bootstrap.childHandler(new ChannelInitializer<Channel>(){

            @Override
            protected void initChannel(Channel channel) throws Exception {
                ByteBuf delimiter = Unpooled.copiedBuffer(TcpConstants.DEFAULT_DECODE_CHAR.getBytes());
                channel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024 * 8, delimiter));

                // 添加编解码器
                channel.pipeline().addLast(new TcpMsgDecoder());
                channel.pipeline().addLast(new TcpMsgEncoder());
                // 添加handler
                EventBus eventBus = new EventBus("nameserver主事件监听器");
                channel.pipeline().addLast(new TcpNettyServerHandler(eventBus));
            }
        });

        // 监听jvm的关闭来关闭线程组
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }));

        // 启动
        ChannelFuture channelFuture = bootstrap.bind(port).sync();
        log.info("start nameserver application on port:{}", port);

        channelFuture.channel().closeFuture().sync();   // sync: 同步关闭, 即阻塞当前线程, 直到channel关闭
        log.info("nameserver is closed");
    }

    /**
     * 启动复制任务
     */
    private void initReplication() {
        //获取到了集群复制的配置属性
        replicationService = new ReplicationService();

        ReplicationModeEnum replicationMode = replicationService.checkProperties();
        replicationService.startReplicationTask(replicationMode);       // 启动连接或者连接主节点

        if (replicationMode == ReplicationModeEnum.MASTER_SLAVE) {
            String role = CommonCache.getNameserverProperties().getMasterSlaveReplicationProperties().getRole();
            ReplicationRoleEnum replicationRole = ReplicationRoleEnum.of(role);

            ReplicationTask replicationTask = null;
            if (replicationRole == ReplicationRoleEnum.MASTER) {
                // 开启同步逻辑
                replicationTask = new MasterReplicationMsgSendTask("同步任务-主节点-消息发送任务");
            } else {
                // 开启心跳任务, 发送给主节点
                replicationTask = new SlaveReplicationHeartBeatTask("同步任务-从节点-心跳任务任务");
            }
            replicationTask.startTaskAsync();
            CommonCache.setReplicationTask(replicationTask);
        } else if (replicationMode == ReplicationModeEnum.TRACE) {
            // 启用链路同步任务
            log.error("链路复制功能暂未实现");
        }
        //master-slave 复制 ？ trace 复制？
        //如果是主从复制-》master角色-》开启一个额外的netty进程-》slave链接接入-》当数据写入master的时候，把写入的数据同步给到slave节点
        //如果是主从复制-》slave角色-》开启一个额外的netty进程-》slave端去链接master节点
    }

    /**
     * 启动超时检测任务
     */
    private void startTimeoutRemovalTask() {
        // 启动超时检测任务
        Thread thread = new Thread(new InValidServiceRemoveTask());
        thread.setName("服务超时检测线程");
        thread.start();
    }
}
